package spring.cloud.tasks.executor.listener.sharding;

import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.api.CuratorWatcher;
import org.apache.curator.framework.state.ConnectionState;
import org.apache.curator.framework.state.ConnectionStateListener;
import spring.cloud.tasks.executor.listener.ListenerHolder;
import spring.cloud.tasks.executor.task.TaskExecutor;
import spring.cloud.tasks.executor.task_type.TaskType;
import spring.cloud.tasks.executor.utils.ThreadFactory;
import spring.cloud.tasks.executor.utils.log.LogUtils;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * 分片监听管理器.
 */
@Slf4j
@Setter
@Getter
public class ShardingListenerHolder extends ListenerHolder {
    public volatile boolean isShutdown;

    private CuratorWatcher necessaryCuratorWatcher;

    private ShardingService shardingService;

    private ExecutorService executorService;

    private ConnectionStateListener connectionStateListener;

    public ShardingListenerHolder(final TaskExecutor taskExecutor) {
        super(taskExecutor);
        this.shardingService = taskExecutor.getShardingService();
        // because cron/passive job do nothing in onResharding method, no need this watcher.
        TaskType taskType = taskExecutor.getTaskConfigurationService().getTaskType();
        if (!taskType.isCron() && !taskType.isPassive()) {
            necessaryCuratorWatcher = new NecessaryCuratorWatcher(this);
        }
    }

    @Override
    public void start() {
        if (necessaryCuratorWatcher != null) {
            executorService = Executors.newSingleThreadExecutor(new ThreadFactory(executorId + "-" + taskId + "-registerNecessaryWatcher", false));
            shardingService.registerNecessaryWatcher(necessaryCuratorWatcher);
            connectionStateListener = new ConnectionStateListener() {
                @Override
                public void stateChanged(CuratorFramework curatorFramework, ConnectionState connectionState) {
                    if ((connectionState == ConnectionState.CONNECTED) || (connectionState == ConnectionState.RECONNECTED)) {
                        // maybe node data have changed, so doBusiness whatever, it's okay for MSG job
                        LogUtils.info(log, taskId, "state change to {}, trigger doBusiness and register necessary watcher.", connectionState);
                        doBusiness();
                        registerNecessaryWatcher();
                    }
                }
            };
            addConnectionStateListener(connectionStateListener);
        }
    }

    @Override
    public void stop() {
        super.stop();
        if (executorService != null) {
            executorService.shutdownNow();
        }
        if (connectionStateListener != null) {
            removeConnectionStateListener(connectionStateListener);
        }
        isShutdown = true;
    }

    public void registerNecessaryWatcher() {
        executorService.execute(new Runnable() {
            @Override
            public void run() {
                if (!isShutdown) {
                    shardingService.registerNecessaryWatcher();
                }
            }
        });
    }

    public void doBusiness() {
        try {
            // cannot block reconnected thread or re-registerNecessaryWatcher, so use thread pool to do business,
            // and the thread pool is the same with job-tree-cache's
            nodeCacheOrTreeCacheManager.getExecutorService().execute(new Runnable() {
                @Override
                public void run() {
                    try {
                        if (isShutdown) {
                            return;
                        }
                        if (taskExecutor == null || taskExecutor.getTaskExecuteEngine() == null) {
                            return;
                        }
                        LogUtils.info(log, taskId, "{} trigger on-resharding", taskId);
                        taskExecutor.getTaskExecuteEngine().onResharding();
                    } catch (Throwable t) {
                        LogUtils.error(log, taskId, "Exception throws during resharding", t);
                    }
                }
            });
        } catch (Throwable t) {
            LogUtils.error(log, taskId, "Exception throws during execute thread", t);
        }
    }

}
